﻿#pragma once
#include <map>
#include <thread>
#include <future>
#include <memory>
#include <queue>
#include <functional>
#include <atomic>
#include <queue>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <stack>
#include <iostream>
#include <cstdarg>
#include <list>

#define DISALLOW_COPY(TypeName) \
   TypeName(const TypeName&)

#define DISALLOW_MOVE(TypeName) \
  void operator=(const TypeName&&)

#define DISALLOW_ASSIGN(TypeName) \
  void operator=(const TypeName&)

#define DISALLOW_COPY_AND_ASSIGN(TypeName) \
  DISALLOW_COPY(TypeName);                 \
  DISALLOW_ASSIGN(TypeName)


#define DISALLOW_MOVE_AND_ASSIGN(TypeName) \
  DISALLOW_MOVE(TypeName);                 \
  DISALLOW_ASSIGN(TypeName)

#define DISALLOW_COPY_AND_MOVE_AND_ASSIGN(TypeName) \
  DISALLOW_COPY(TypeName);                          \
  DISALLOW_MOVE(TypeName);                          \
  DISALLOW_ASSIGN(TypeName)

#define DEBUGPRINT(ARG) std::cout << ARG << std::endl

// a barrier nem felelős a notify-ért, azt oldja meg a joiner
// a separator nem működik jól.
// pointerek kellenek és használjuk a share_ptr.-t
//atomic type-oknál utálom a pointert igazából azért is lett eldöntve, hogy ne ez legyen.
namespace dvo{

	class spin_mutex{
	private:
		mutable std::atomic_flag flag_;
		DISALLOW_COPY_AND_MOVE_AND_ASSIGN(spin_mutex);
	public:
		spin_mutex() { flag_.clear(); }
		~spin_mutex() { }
		void lock() const { while (flag_.test_and_set()); }
		void unlock() const { flag_.clear(); }
	};
	struct spin_guard{
		const spin_mutex& spin_;
		spin_guard(const spin_mutex& spin) : spin_(spin) { spin_.lock(); }
		~spin_guard(){ spin_.unlock(); }
	private:
		DISALLOW_COPY_AND_MOVE_AND_ASSIGN(spin_guard);
	};

	struct flag_guard{
		flag_guard(std::atomic_flag& flag) : flag_(flag) { flag_.test_and_set(); }
		~flag_guard() { flag_.clear(); }
	private:
		std::atomic_flag& flag_;
	};

	template<typename T>
	struct promise_setter{
		explicit promise_setter(std::promise<T>& promise, T& value) : promise_(promise), value_(value) {}
		~promise_setter() { promise_.set_value(value_); }
	private:
		std::promise<T>& promise_;
		T& value_;
	};

	namespace ps
	{
		template<typename T> class ISimplex{
		public:
			virtual void Transmit(T*) = 0;
			virtual void Transmit(T&) = 0;
			virtual void Transmit(T&&) = 0;
		};
		template<typename T>
		T& operator>>(T& value, ISimplex<T>& target)
		{
			target.Transmit(value);
			return value;
		}
		template<typename T>
		void operator>>(T&& value, ISimplex<T>& target)
		{
			target.Transmit(value);
		}
		template<typename T> class Socket : public ISimplex < T > {
		public:
			Socket() {}
			Socket(Socket<T>& peer) { Bind(peer.transmit_); }
			Socket(Socket<T>&& peer) { Bind(peer.transmit_); }
			Socket(std::function<void(T&)>&& func) { Bind(func); }
			virtual ~Socket() {}
			Socket<T>& operator=(const std::function<void(T&)>& func) { Bind(func); return *this; }
			void Bind(const std::function<void(T&)>& func){ transmit_ = func; }
			virtual void Transmit(T* value) override { transmit_(*value); }
			virtual void Transmit(T& value) override { transmit_(value); }
			virtual void Transmit(T&& value) override { transmit_(value); }
		private:
			std::function<void(T&)> transmit_;
		};
		template<typename T>
		void operator>>(Socket<T>& socket, const std::function<void(T&)>& func){
			socket.Bind(func);
		}
		template<typename TKey, typename TValue> class KeyedSocket : public ISimplex < TValue > {
		public:
			KeyedSocket(TKey key) : key_(key) { }
			KeyedSocket(KeyedSocket<TKey, TValue>& peer) : key_(peer.key_), transmit_(peer.transmit_) { }
			virtual ~KeyedSocket() { }
			void Bind(const std::function<void(TKey, TValue&)>& func){ transmit_ = func; }
			virtual void Transmit(TValue* value) override { transmit_(key_, *value); }
			virtual void Transmit(TValue& value) override { transmit_(key_, value); }
			virtual void Transmit(TValue&& value) override { transmit_(key_, value); }
		private:
			TKey key_;
			std::function<void(TKey, TValue&)> transmit_;
		};
		template<typename TKey, typename TValue>
		void operator>>(KeyedSocket<TKey, TValue>& socket, const std::function<void(TValue&)>& func){
			socket.Bind(func);
		}
		template<typename T> class Plug : public ISimplex < T > {
		public:
			Plug() { }
			Plug(Plug<T>& peer) : target_(peer.target_) { }
			virtual ~Plug() { }
			inline void Connect(ISimplex<T>* target) { target_ = target; }
			virtual void Transmit(T* value) override { target_->Transmit(*value); }
			virtual void Transmit(T& value) override { target_->Transmit(value); }
			virtual void Transmit(T&& value) override { target_->Transmit(value); }
		private:
			ISimplex<T>* target_;
		};
        template<typename T>
		Plug<T>& operator>>(Plug<T>& plug, Plug<T>& target){
			plug.Connect(&target);
			return target;
		}
		template<typename T>
		Plug<T>& operator>>(Plug<T>& plug, Plug<T>* target){
			plug.Connect(target);
			return *target;
		}
		template<typename T>
		void operator>>(Plug<T>& plug, ISimplex<T>* target){
			plug.Connect(target);
		}
        template<typename T>
		void operator>>(Plug<T>& plug, ISimplex<T>& target){
			plug.Connect(&target);
		}
		//---------------------------------------------------------------------------------

		template<typename T> class Receiver {
		public:
			Receiver();
			virtual ~Receiver();
			Socket<T>& Input(){ return input_; }
		protected:
			virtual void Process(T& value) = 0;
		private:
			Socket<T> input_;
		};

		template<typename T>
		Receiver<T>::Receiver()
		{
			input_.Bind(std::bind(&Receiver<T>::Process, this, std::placeholders::_1));
		}

		template<typename T>
		Receiver<T>::~Receiver()
		{

		}
        //--------------------------------------------------------------------------------
        template<typename T>
		Receiver<T>& operator>>(Plug<T>& source, Receiver<T>& target)
		{
            source >> target.Input();
			return target;
		}
		//---------------------------------------------------------------------------------
		template<typename T>
		class SimpleReceiver : public Receiver<T>
		{
		public:
			SimpleReceiver(std::function<void(T&)> process) { process_ = process; }
			SimpleReceiver(SimpleReceiver<T>& peer) : process_(peer.process_) { }
			SimpleReceiver(SimpleReceiver<T>&& peer) : process_(peer.process_) { }
			virtual ~SimpleReceiver(){}
			SimpleReceiver& operator=(std::function<void(T&)> process) { process_ = process; return *this; }
		protected:
			virtual void Process(T& value) override { process_(value); }
		private:
			std::function<void(T&)> process_;
		};

		//---------------------------------------------------------------------------------

		template<typename T> class Transmitter {
		public:
			Transmitter();
			virtual ~Transmitter();
			Plug<T>& Output(){ return output_; }
			virtual void Process() = 0;
		protected:
			inline void Send(T& value) { output_.Transmit(value); }
		private:
			Plug<T> output_;
		};

		template<typename T>
		Transmitter<T>::Transmitter()
		{
			
		}

		template<typename T>
		Transmitter<T>::~Transmitter()
		{

		}
        
		//---------------------------------------------------------------------------------
		template<typename T>
		class SimpleTransmitter : public Transmitter<T>
		{
		public:
			SimpleTransmitter(std::function<T&(void)> process) { process_ = process; }
			SimpleTransmitter(SimpleTransmitter<T>& peer) : process_(peer.process_) { }
			SimpleTransmitter(SimpleTransmitter<T>&& peer) : process_(peer.process_) { }
			virtual ~SimpleTransmitter() { }
			SimpleTransmitter& operator=(std::function<T&(void)> process) { process_ = process; return *this; }
			virtual void Process() override { Send(process_()); }
		private:
			std::function<T&(void)> process_;
		};

		//---------------------------------------------------------------------------------

		template<typename TInput, typename TOutput> class Transceiver {
		public:
			Transceiver();
			virtual ~Transceiver();
			Socket<TInput>& Input(){ return input_; }
			Plug<TOutput>& Output(){ return output_; }
		protected:
			virtual void Process(TInput& value) = 0;
			inline void Send(TOutput& value) { output_.Transmit(value); }
			inline void Send(TOutput&& value) { output_.Transmit(value); }
		private:
			Socket<TInput> input_;
			Plug<TOutput> output_;
		};

		template<typename TInput, typename TOutput>
		Transceiver<TInput, TOutput>::Transceiver()
		{
			input_.Bind(std::bind(&Transceiver<TInput, TOutput>::Process, this, std::placeholders::_1));
		}

		template<typename TInput, typename TOutput>
		Transceiver<TInput, TOutput>::~Transceiver()
		{

		}

		//---------------------------------------------------------------------------------
		template<typename TInput, typename TOutput>
		class SimpleTransceiver : public Transceiver<TInput, TOutput>
		{
		public:
			SimpleTransceiver& operator=(std::function<TOutput&(TInput&)> process) { process_ = process; return *this; }
			SimpleTransceiver(std::function<TOutput&(TInput&)> process) { process_ = process; }
			SimpleTransceiver(SimpleTransceiver<TInput, TOutput>& peer) : process_(peer.process_) { }
			SimpleTransceiver(SimpleTransceiver<TInput, TOutput>&& peer) : process_(peer.process_) { }
			virtual ~SimpleTransceiver(){}
		protected:
			virtual void Process(TInput& value) override { Send(process_(value)); }
		private:
			std::function<TOutput&(TInput&)> process_;
		};

		//---------------------------------------------------------------------------------

		template<typename T>
		class Capacitor : public Transceiver<T, T>{
		public:
			Capacitor(int capacity = 1) : capacity_(capacity) {}
			inline void SetCapacity(int capacity) { capacity_ = capacity; }
			inline int& GetCapacity() { return capacity_; }
		protected:
			virtual void Process(T& item) override;
		private:
			unsigned int capacity_;
			std::list<T*> items_;
		};

		template<typename T>
		void Capacitor<T>::Process(T& item) {
			items_.push_back(&item);
			if (items_.size() < capacity_){
				return;
			}
			while (!items_.empty()){
				T* item = items_.front();
				items_.pop_back();
				Send(*item);
			}
		}

		//---------------------------------------------------------------------------------

		template<typename TKey, typename TInput, typename TOutput> class Joiner {
		public:
			Joiner();
			virtual ~Joiner();
			KeyedSocket<TKey, TInput>& Inputs(TKey key){ return inputs_.at(key); }
			Plug<TOutput>& Output(){ return output_; }
            inline void Wait() { if (signaled_.test_and_set()) signal_.wait(); }
		protected:
			virtual void Add(TKey key);
			virtual void Remove(TKey key);
			virtual void Process() = 0;
			inline void Send(TOutput& value) { output_.Transmit(value); }
			virtual void Set(TKey key, TInput& value);
			virtual TInput& Get(TKey);
			std::map<TKey, TInput*>& GetAll();
		private:
			std::map<TKey, KeyedSocket<TKey, TInput>> inputs_;
			Plug<TOutput> output_;
			std::map<TKey, TInput*> values_;
			std::condition_variable cv_;
			std::atomic_flag signaled_;
            std::future<void> signal_;
			std::size_t arrived_;
			std::size_t limit_;
			std::mutex mutex_;
			std::size_t left_;
		};

		template<typename TKey, typename TInput, typename TOutput>
		Joiner<TKey, TInput, TOutput>::Joiner() : limit_(0), left_(0), arrived_(0)
		{
            signaled_.clear();
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Joiner<TKey, TInput, TOutput>::Add(TKey key)
		{
			++limit_;
			inputs_.emplace(key, key);
			inputs_.at(key).Bind(std::bind(&Joiner<TKey, TInput, TOutput>::Set, this, std::placeholders::_1, std::placeholders::_2));
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Joiner<TKey, TInput, TOutput>::Remove(TKey key)
		{
			--limit_;
			inputs_.erase(key);
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Joiner<TKey, TInput, TOutput>::Set(TKey key, TInput& value)
		{
			std::unique_lock<std::mutex> mutex(mutex_);
			if (++arrived_ >= limit_){
				arrived_ = 0;
				cv_.notify_all();
			}
			else{
				cv_.wait(mutex);
			}
			values_[key] = &value;
			if (++left_ < limit_){
				return;
			}
			left_ = 0;
			signal_ = std::async(std::launch::async, [&](){
				std::lock_guard<std::mutex> lock(mutex_);
				Process();
			});
            signaled_.test_and_set();
		}

		template<typename TKey, typename TInput, typename TOutput>
		TInput& Joiner<TKey, TInput, TOutput>::Get(TKey key)
		{
			return *values_[key];
		}

		template<typename TKey, typename TInput, typename TOutput>
		std::map<TKey, TInput*>& Joiner<TKey, TInput, TOutput>::GetAll()
		{
			return values_;
		}

		template<typename TKey, typename TInput, typename TOutput>
		Joiner<TKey, TInput, TOutput>::~Joiner()
		{

		}
		//---------------------------------------------------------------------------------
		template<typename TKey, typename TInput, typename TOutput>
		class SimpleJoiner : public Joiner < TKey, TInput, TOutput > {
		public:
			SimpleJoiner(TKey key1, TKey key2){ Add(key1); Add(key2); }
			SimpleJoiner(TKey key1, TKey key2, TKey key3){ Add(key1); Add(key2); Add(key3); }
			SimpleJoiner(TKey key1, TKey key2, TKey key3, TKey key4){ Add(key1); Add(key2); Add(key3); Add(key4); }
			SimpleJoiner(TKey key1, TKey key2, TKey key3, TKey key4, TKey key5){ Add(key1); Add(key2); Add(key3); Add(key4); Add(key5); }
			virtual ~SimpleJoiner() { }
			SimpleJoiner& operator=(const std::function<TOutput&(const std::map<TKey, TInput*>&)>& process) { process_ = process; return *this; }
		protected:
			virtual void Process() override { Send(process_(GetAll())); }
		private:
			SimpleJoiner(SimpleJoiner<TKey, TInput, TOutput>& peer) { }
			std::function<TOutput&(const std::map<TKey, TInput*>&)> process_;
		};
		//---------------------------------------------------------------------------------

		template<typename TKey, typename TInput, typename TOutput> class Detacher {
		public:
			Detacher();
			virtual ~Detacher();
			Socket<TInput>& Input(){ return input_; }
			Plug<TOutput>& Outputs(TKey key){ return outputs_.at(key); }
		protected:
			virtual void Add(TKey key);
			virtual void Remove(TKey key);
			virtual std::future<void> Send(TKey key, TOutput& value);
			virtual void Process(TInput& value) = 0;
		private:
			Socket<TInput> input_;
			std::map<TKey, Plug<TOutput>> outputs_;
		};

		template<typename TKey, typename TInput, typename TOutput>
		Detacher<TKey, TInput, TOutput>::Detacher()
		{
			input_.Bind(std::bind(&Detacher<TKey, TInput, TOutput>::Process, this, std::placeholders::_1));
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Detacher<TKey, TInput, TOutput>::Add(TKey key)
		{
			outputs_.emplace(key, Plug<TOutput>());
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Detacher<TKey, TInput, TOutput>::Remove(TKey key)
		{
			outputs_.erase(key);
		}

		template<typename TKey, typename TInput, typename TOutput>
		std::future<void> Detacher<TKey, TInput, TOutput>::Send(TKey key, TOutput& value)
		{
			return std::async(std::launch::async, [&](TKey fwdK, TOutput& fwdV) { outputs_[fwdK].Transmit(fwdV); }, key, std::ref(value));
		}

		template<typename TKey, typename TInput, typename TOutput>
		Detacher<TKey, TInput, TOutput>::~Detacher()
		{

		}

		//---------------------------------------------------------------------------------
		template<typename TKey, typename TInput, typename TOutput>
		class SimpleDetacher : public Detacher < TKey, TInput, TOutput > {
		public:
			SimpleDetacher(TKey key1, TKey key2){ Add(key1); Add(key2); }
			SimpleDetacher(TKey key1, TKey key2, TKey key3){ Add(key1); Add(key2); Add(key3); }
			SimpleDetacher(TKey key1, TKey key2, TKey key3, TKey key4){ Add(key1); Add(key2); Add(key3); Add(key4); }
			SimpleDetacher(TKey key1, TKey key2, TKey key3, TKey key4, TKey key5){ Add(key1); Add(key2); Add(key3); Add(key4); Add(key5); }
			SimpleDetacher& operator=(const std::function<const std::map<TKey, TOutput&>(TInput&)> process) { process_ = process; return *this; }
			virtual ~SimpleDetacher(){}
		protected:
			virtual void Process(TInput& value) override { for (auto pair : process_(value)) { Send(pair.first, pair.second); } }
		private:
			SimpleDetacher(SimpleDetacher<TKey, TInput, TOutput>& peer) { }
			std::function<const std::map<TKey, TOutput&>(TInput&)> process_;
		};
		//---------------------------------------------------------------------------------

		template<typename TKey, typename TInput, typename TOutput> class Selector {
		public:
			Selector();
			virtual ~Selector();
			Socket<TInput>& Input(){ return input_; }
			Plug<TOutput>& Outputs(TKey key){ return outputs_.at(key); }
		protected:
			Selector(Selector<TKey, TInput, TOutput>& peer) {  }
			virtual void Add(TKey key);
			inline void Send(TKey key, TOutput& value) { outputs_.at(key).Transmit(value); }
			virtual void Process(TInput& value) = 0;
		private:
			Socket<TInput> input_;
			std::map<TKey, Plug<TOutput>> outputs_;
		};

		template<typename TKey, typename TInput, typename TOutput>
		Selector<TKey, TInput, TOutput>::Selector()
		{
			input_.Bind(std::bind(&Selector<TKey, TInput, TOutput>::Process, this, std::placeholders::_1));
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Selector<TKey, TInput, TOutput>::Add(TKey key)
		{
			outputs_.emplace(key, Plug<TOutput>());
		}

		template<typename TKey, typename TInput, typename TOutput>
		Selector<TKey, TInput, TOutput>::~Selector()
		{

		}
		//---------------------------------------------------------------------------------
		template<typename TKey, typename TInput, typename TOutput>
		class SimpleSelector : public Selector < TKey, TInput, TOutput > {
		public:
			SimpleSelector(TKey key1, TKey key2){ Add(key1); Add(key2); }
			SimpleSelector(TKey key1, TKey key2, TKey key3){ Add(key1); Add(key2); Add(key3); }
			SimpleSelector(TKey key1, TKey key2, TKey key3, TKey key4){ Add(key1); Add(key2); Add(key3); Add(key4); }
			SimpleSelector(TKey key1, TKey key2, TKey key3, TKey key4, TKey key5){ Add(key1); Add(key2); Add(key3); Add(key4); Add(key5); }
			//SimpleSelector& operator=(const std::function<TOutput&(TInput&)>& process) { process_ = process; return *this; }
			SimpleSelector& operator=(const TKey& key) { key_ = key; return *this; }
			Plug<TOutput>& operator()(const TKey& key) { return Outputs(key); }
			void operator()(const std::function<TOutput&(TInput&)>& process) { process_ = process; }
			virtual ~SimpleSelector(){}
		protected:
			virtual void Process(TInput& value) override { Send(key_, process_(value)); }
		private:
			SimpleSelector(SimpleSelector<TKey, TInput, TOutput>& peer) {  }
			TKey key_;
			std::function<TOutput&(TInput&)> process_;
		};

		//---------------------------------------------------------------------------------

		template<typename TKey, typename TInput, typename TOutput> class Restrictor {
		public:
			Restrictor();
			virtual ~Restrictor();
			KeyedSocket<TKey, TInput>& Inputs(TKey key){ return inputs_.at(key); }
			Plug<TOutput>& Output(){ return output_; }
		protected:
			Restrictor(Restrictor<TKey, TInput, TOutput>& peer) {  }
			virtual void Add(TKey key);
			virtual void Process(TKey, TInput&) = 0;
			inline void Send(TOutput& value) { output_.Transmit(value); }
		private:
			std::map<TKey, KeyedSocket<TKey, TInput>> inputs_;
			Plug<TOutput> output_;
		};

		template<typename TKey, typename TInput, typename TOutput>
		Restrictor<TKey, TInput, TOutput>::Restrictor()
		{

		}

		template<typename TKey, typename TInput, typename TOutput>
		void Restrictor<TKey, TInput, TOutput>::Add(TKey key)
		{
			inputs_.emplace(key, key);
			inputs_.at(key).Bind(std::bind(&Restrictor<TKey, TInput, TOutput>::Process, this, std::placeholders::_1, std::placeholders::_2));
		}

		template<typename TKey, typename TInput, typename TOutput>
		Restrictor<TKey, TInput, TOutput>::~Restrictor()
		{

		}
		//---------------------------------------------------------------------------------
		template<typename TKey, typename TInput, typename TOutput>
		class SimpleRestrictor : public Restrictor < TKey, TInput, TOutput > {
		public:
			SimpleRestrictor(TKey key1, TKey key2){ Add(key1); Add(key2); }
			SimpleRestrictor(TKey key1, TKey key2, TKey key3){ Add(key1); Add(key2); Add(key3); }
			SimpleRestrictor(TKey key1, TKey key2, TKey key3, TKey key4){ Add(key1); Add(key2); Add(key3); Add(key4); }
			SimpleRestrictor(TKey key1, TKey key2, TKey key3, TKey key4, TKey key5){ Add(key1); Add(key2); Add(key3); Add(key4); Add(key5); }
			//SimpleRestrictor& operator=(const std::function<TOutput*(TKey, TInput&)>& process) { process_ = process; return *this; }
			SimpleRestrictor& operator=(const TKey& key) { key_ = key;  return *this; }
			KeyedSocket<TKey, TInput>& operator()(const TKey& key) { return Inputs(key); }
			void operator()(const std::function<TOutput&(TInput&)>& process) { process_ = process; }
			virtual ~SimpleRestrictor(){}
		protected:
			virtual void Process(TKey key, TInput& value) override { if (key_ != key) return; Send(process_(value)); }
		private:
			SimpleRestrictor(SimpleRestrictor<TKey, TInput, TOutput>& peer) {  }
			std::function<TOutput&(TInput&)> process_;
			TKey key_;
		};

		//---------------------------------------------------------------------------------

		template<typename T> class Separator {
		public:
			Separator();
			virtual ~Separator();
			Socket<T>& Input(){ return input_; }
			Plug<T>& Output(){ return output_; }
			inline void Wait() { signal_.wait(); }
		protected:
			Separator(Separator<T>& peer) { }
			virtual void Receive(T& value);
			inline void Send(T& value) { output_.Transmit(value); }
		private:
			Socket<T> input_;
			Plug<T> output_;
			std::future<void> signal_;
			std::mutex mutex_;
			T* value_;
		};

		template<typename T>
		Separator<T>::Separator()
		{
			value_ = nullptr;
			input_.Bind(std::bind(&Separator<T>::Receive, this, std::placeholders::_1));
		}

		template<typename T>
		void Separator<T>::Receive(T& value)
		{
			std::lock_guard<std::mutex> lock(mutex_);
			value_ = &value;
			signal_ = std::async(std::launch::async, [&]()
			{
				std::unique_lock<std::mutex> mutex(mutex_);
				T& forwarded = *value_;
				Send(forwarded);
			});
		}

		template<typename T>
		Separator<T>::~Separator()
		{

		}
	}//end of Pushflow namespace


	namespace pl
	{
		template<typename T> class ISimplex{
		public:
			virtual T& Request() = 0;
		};
		template<typename T>
		T& operator<<(ISimplex<T>& source, T& target){
			target = source.Request();
			return target;
		}
		template<typename T>
		T* operator<<(ISimplex<T>& source, T*& target){
			T& result = source.Request();
            target = &result;
			return target;
		}
		template<typename T>
		T& operator<<(T& target, ISimplex<T>& source){
			target = source.Request();
			return target;
		}
		template<typename T> class Socket : public ISimplex < T > {
		public:
			Socket() {}
			Socket(Socket<T>& peer) { Bind(peer.transmitter_); }
			Socket(Socket<T>&& peer) { Bind(peer.requester_); }
			Socket(std::function<T&(void)>&& func) { Bind(func); }
			Socket<T>& operator=(const std::function<T&(void)>& func) { Bind(func); return *this; }
			void Bind(const std::function<T&(void)>& func){ requester_ = func; }
			virtual T& Request() override { return requester_(); }
		private:
			std::function<T&(void)> requester_;
		};
		template<typename TKey, typename TValue> class KeyedSocket : public ISimplex < TValue > {
		public:
			KeyedSocket(TKey key) : key_(key) { }
			KeyedSocket(KeyedSocket<TKey, TValue>& peer) : key_(peer.key_), request_(peer.request_) { }
			void Bind(std::function<TValue&(TKey)> func){ request_ = func; }
			virtual TValue& Request() override { return request_(key_); }
		private:
			TKey key_;
			std::function<TValue&(TKey)> request_;
		};
		template<typename TKey, typename TValue>
		void operator<<(KeyedSocket<TKey, TValue>& socket, const std::function<TValue&(void)>& func){
			socket.Bind(func);
		}
		template<typename T> class Plug : public ISimplex<T> {
		public:
			Plug() { }
			Plug(Plug<T>& peer) : target_(peer.target_) { }
			inline void Connect(ISimplex<T>* target) { target_ = target; }
			virtual T& Request() override { return target_->Request(); }
		private:
			ISimplex<T>* target_;
		};
		template<typename T>
		Plug<T>& operator<<(Socket<T>& source, Plug<T>& target){
			target.Connect(&source);
			return target;
		}

        template<typename T>
        Plug<T>& operator>>=(Socket<T>& source, Plug<T>& target){
			target.Connect(&source);
			return target;
		}
		//---------------------------------------------------------------------------------

		template<typename T> class Receiver {
		public:
			Receiver();
			virtual ~Receiver();
			Plug<T>& Input(){ return input_; }
		protected:
			Receiver(Receiver<T>& peer) { }
			virtual void Process() = 0;
			inline T& Demand() { return input_.Request(); }
		private:
			Plug<T> input_;
		};

		template<typename T>
		Receiver<T>::Receiver()
		{

		}

		template<typename T>
		Receiver<T>::~Receiver()
		{

		}
		//---------------------------------------------------------------------------------
		template<typename T>
		class SimpleReceiver : public Receiver<T>
		{
		public:
			SimpleReceiver& operator=(const std::function<void(T&)>& process) { process_ = process; return *this; }
			SimpleReceiver(const std::function<void(T&)> process) { process_ = process; }
			virtual ~SimpleReceiver(){}
			void operator()(void) { Process(); }
			virtual void Process() override { process_(Demand()); }
		private:
			std::function<void(T&)> process_;
		};
		//---------------------------------------------------------------------------------
		template<typename T> class Transmitter {
		public:
			Transmitter();
			virtual ~Transmitter();
			Socket<T>& Output(){ return output_; }
		protected:
			Transmitter(Transmitter<T>& peer) { }
			virtual T& Process() = 0;
		private:
			Socket<T> output_;
		};

		template<typename T>
		Transmitter<T>::Transmitter()
		{
			output_.Bind(std::bind(&Transmitter<T>::Process, this));
		}

		template<typename T>
		Transmitter<T>::~Transmitter()
		{

		}
		//---------------------------------------------------------------------------------
		template<typename T>
		class SimpleTransmitter : public Transmitter<T>
		{
		public:
			SimpleTransmitter(SimpleTransmitter<T>& peer) : process_(peer.process_) { }
			SimpleTransmitter& operator=(const std::function<T&(void)>& process) { process_ = process; return *this; }
			SimpleTransmitter(const std::function<T&(void)>& process) { process_ = process; }
			virtual ~SimpleTransmitter(){}
		protected:
			virtual T& Process() override { return process_(); }
		private:
			std::function<T&(void)> process_;
		};
		//---------------------------------------------------------------------------------
		template<typename TInput, typename TOutput> class Transceiver {
		public:
			Transceiver();
			virtual ~Transceiver();
			Plug<TInput>& Input(){ return input_; }
			Socket<TOutput>& Output(){ return output_; }
		protected:
			Transceiver(Transceiver<TInput, TOutput>& peer) { }
			inline TInput& Demand() { return input_.Request(); }
			virtual TOutput& Process() = 0;
		private:
			Plug<TInput> input_;
			Socket<TOutput> output_;
		};

		template<typename TInput, typename TOutput>
		Transceiver<TInput, TOutput>::Transceiver()
		{
			output_.Bind(std::bind(&Transceiver<TInput, TOutput>::Process, this));
		}

		template<typename TInput, typename TOutput>
		Transceiver<TInput, TOutput>::~Transceiver()
		{

		};
		//---------------------------------------------------------------------------------
		template<typename TInput, typename TOutput>
		class SimpleTransceiver : public Transceiver<TInput, TOutput>
		{
		public:
			SimpleTransceiver(SimpleTransceiver<TInput, TOutput>& peer) : process_(peer.process_) { }
			SimpleTransceiver& operator=(const std::function<TOutput&(TInput&)>& process) { process_ = process; return *this; }
			SimpleTransceiver(const std::function<TOutput&(TInput&)>& process) { process_ = process; }
			virtual ~SimpleTransceiver(){}
		protected:
			virtual TOutput& Process() override { return process_(Demand()); }
		private:
			std::function<TOutput&(TInput&)> process_;
		};
		//---------------------------------------------------------------------------------

		template<typename T>
		class Capacitor : public Transceiver<T, T>{
		public:
			Capacitor(int capacity = 1) : capacity_(capacity) {}
			inline void SetCapacity(int capacity) { capacity_ = capacity; }
			inline int& GetCapacity() { return capacity_; }
		protected:
			virtual T& Process() override;
		private:
			inline T& Emit() { T* item = items_.front(); items_.pop(); return *item; }
			unsigned int capacity_;
			std::list<T*> items_;
		};

		template<typename T>
		T& Capacitor<T>::Process() {
			if (!items_.empty()){
				return Emit();
			}
			for (int i = 0; i < limit_; ++i){
				T& item = Demand();
				items_.push_back(&item);
			}
			return Emit();
		}

		//---------------------------------------------------------------------------------

		template<typename TContainer, typename T>
		class Iterator : public Transceiver<TContainer, T>{

		};

		template<typename T>
		class Iterator <std::list<T*>, T > : public Transceiver<std::list<T*>, T>{
		public:
			Iterator() : items_(nullptr) { }
		protected:
			T& Process() override;
		private:
			std::list<T*>* items_;
		};

		template<typename T>
		T& Iterator<std::list<T*>, T>::Process()
		{
			if (items_ == nullptr || items_->empty()){
				items_ = &Demand();
			}
			T& result = *items_->front();
			items_->pop_back();
			return result;
		}
		//---------------------------------------------------------------------------------

		template<typename TKey, typename TInput, typename TOutput> class Joiner {
		public:
			Joiner();
			virtual ~Joiner();
			Plug<TInput>& Input(){ return input_; }
			KeyedSocket<TKey, TOutput>& Outputs(TKey key){ return outputs_.at(key); }
		protected:
			Joiner(Joiner<TKey, TInput, TOutput>& peer) { }
			virtual void Add(TKey key);
			virtual void Remove(TKey key);
			virtual TOutput& Supply(TKey key);
			inline TInput& Demand() { return input_.Request(); }
			virtual TOutput& Process(TKey key, TInput& value) = 0;
		private:
			std::map<TKey, KeyedSocket<TKey, TOutput>> outputs_;
			Plug<TInput> input_;
			std::condition_variable cv_;
			std::size_t arrived_;
			std::size_t limit_;
			std::mutex mutex_;

		};

		template<typename TKey, typename TInput, typename TOutput>
		Joiner<TKey, TInput, TOutput>::Joiner() : arrived_(0), limit_(0)
		{
			
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Joiner<TKey, TInput, TOutput>::Add(TKey key)
		{
			outputs_.emplace(key, key);
			outputs_.at(key).Bind(std::bind(&Joiner<TKey, TInput, TOutput>::Supply, this, std::placeholders::_1));
			++limit_;
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Joiner<TKey, TInput, TOutput>::Remove(TKey key)
		{
			outputs_.erase(key);
			--limit_;
		}


		template<typename TKey, typename TInput, typename TOutput>
		TOutput& Joiner<TKey, TInput, TOutput>::Supply(TKey key)
		{
			std::unique_lock<std::mutex> lock(mutex_);
			TInput& value = Demand();
			if (++arrived_ < limit_){
				cv_.wait(lock);
			}
			else{
				cv_.notify_all();
				arrived_ = 0;
			}
			TOutput& result = Process(key, value);
			return result;
		}

		template<typename TKey, typename TInput, typename TOutput>
		Joiner<TKey, TInput, TOutput>::~Joiner()
		{

		}
		//---------------------------------------------------------------------------------
		template<typename TKey, typename TInput, typename TOutput>
		class SimpleJoiner : public Joiner< TKey, TInput, TOutput > {
		public:
			SimpleJoiner(TKey key1, TKey key2){ Add(key1); Add(key2); }
			SimpleJoiner(TKey key1, TKey key2, TKey key3){ Add(key1); Add(key2); Add(key3); }
			SimpleJoiner(TKey key1, TKey key2, TKey key3, TKey key4){ Add(key1); Add(key2); Add(key3); Add(key4); }
			SimpleJoiner(TKey key1, TKey key2, TKey key3, TKey key4, TKey key5){ Add(key1); Add(key2); Add(key3); Add(key4); Add(key5); }
            void operator=(const TKey& key) {key_ = key;}
			void operator=(const std::function<TOutput&(TKey, TInput&)>& process) { process_ = process; }
            SimpleJoiner<TKey, TInput, TOutput>& operator()
              (const std::function<TOutput&(TKey, TInput&)>& process) { process_ = process; return *this; }
            KeyedSocket<TKey, TOutput>& operator()(const TKey& key) { return Outputs(key); }
            virtual ~SimpleJoiner(){}
		protected:
			SimpleJoiner(SimpleJoiner<TKey, TInput, TOutput>& peer) { }
			virtual TOutput& Process(TKey key, TInput& value) override { return process_(key, value); }
		private:
			std::function<TOutput&(TKey, TInput&)> process_;
		};

		//---------------------------------------------------------------------------------

		template<typename TKey, typename TInput, typename TOutput> class Detacher {
		public:
			Detacher();
			virtual ~Detacher();
			Plug<TInput>& Inputs(TKey key){ return inputs_.at(key); }
			Socket<TOutput>& Output(){ return output_; }
		protected:
			Detacher(Detacher<TKey, TInput, TOutput>& peer) { }
			virtual void Add(TKey key);
			virtual void Remove(TKey key);
			virtual std::future<TInput&> Demand(TKey key);
			virtual TOutput& Process() = 0;
			std::vector<TKey> GetAll();
		private:
			std::map<TKey, Plug<TInput>> inputs_;
			Socket<TOutput> output_;
		};

		template<typename TKey, typename TInput, typename TOutput>
		Detacher<TKey, TInput, TOutput>::Detacher()
		{
			output_.Bind(std::bind(&Detacher<TKey, TInput, TOutput>::Process, this));
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Detacher<TKey, TInput, TOutput>::Add(TKey key)
		{
			inputs_.emplace(key, Plug<TInput>());
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Detacher<TKey, TInput, TOutput>::Remove(TKey key)
		{
			inputs_.erase(key);
		}

		template<typename TKey, typename TInput, typename TOutput>
		std::vector<TKey> Detacher<TKey, TInput, TOutput>::GetAll()
		{
			std::vector<TKey> result;
			for (std::map<TKey, Plug<TInput>>::iterator it = inputs_.begin(); it != inputs_.end(); ++it)
			{
				result.push_back(it->first);
			}
			return result;
		}

		template<typename TKey, typename TInput, typename TOutput>
		std::future<TInput&> Detacher<TKey, TInput, TOutput>::Demand(TKey key)
		{
			return std::async(std::launch::async, [&](TKey forwarded)->TInput& { return inputs_.at(forwarded).Request(); }, key);
		}

		template<typename TKey, typename TInput, typename TOutput>
		Detacher<TKey, TInput, TOutput>::~Detacher()
		{

		}

		//---------------------------------------------------------------------------------
		template<typename TKey, typename TInput, typename TOutput>
		class SimpleDetacher : public Detacher < TKey, TInput, TOutput > {
		public:
			SimpleDetacher(TKey key1, TKey key2){ Add(key1); Add(key2); }
			SimpleDetacher(TKey key1, TKey key2, TKey key3){ Add(key1); Add(key2); Add(key3); }
			SimpleDetacher(TKey key1, TKey key2, TKey key3, TKey key4){ Add(key1); Add(key2); Add(key3); Add(key4); }
			SimpleDetacher(TKey key1, TKey key2, TKey key3, TKey key4, TKey key5){ Add(key1); Add(key2); Add(key3); Add(key4); Add(key5); }
			void operator=(const std::function<TOutput&(std::map<TKey, TInput*>&)>& process) { process_ = process; }
            Plug<TInput>& operator()(const TKey& key) { return Inputs(key); }
            virtual ~SimpleDetacher(){}
		protected:
			SimpleDetacher(SimpleDetacher<TKey, TInput, TOutput>& peer) { }
			virtual TOutput& Process() override
			{
				std::map<TKey, std::future<TInput&>> futures;
				std::vector<TKey> keys = GetAll();
				for (TKey& key : keys){ futures.emplace(key, Demand(key)); }
				std::map<TKey, TInput*> values;
				for (TKey& key : keys){ futures[key].wait(); values[key] = &futures[key].get(); }
				return process_(values);
			}
		private:
			std::function<TOutput&(std::map<TKey, TInput*>&)> process_;
		};

		//---------------------------------------------------------------------------------

		template<typename TKey, typename TInput, typename TOutput> class Selector {
		public:
			Selector();
			virtual ~Selector();
			Plug<TInput>& Inputs(TKey key){ return inputs_.at(key); }
			Socket<TOutput>& Output(){ return output_; }
		protected:
			Selector(Selector<TKey, TInput, TOutput>& peer) { }
			virtual void Add(TKey key);
			virtual void Remove(TKey key);
			virtual TInput& Demand(TKey key);
			virtual TOutput& Process() = 0;
		private:
			std::map<TKey, Plug<TInput>> inputs_;
			Socket<TOutput> output_;
		};

		template<typename TKey, typename TInput, typename TOutput>
		Selector<TKey, TInput, TOutput>::Selector()
		{
			output_.Bind(std::bind(&Selector<TKey, TInput, TOutput>::Process, this));
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Selector<TKey, TInput, TOutput>::Add(TKey key)
		{
			inputs_.emplace(key, Plug<TInput>());
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Selector<TKey, TInput, TOutput>::Remove(TKey key)
		{
			inputs_.erase(key);
		}

		template<typename TKey, typename TInput, typename TOutput>
		TInput& Selector<TKey, TInput, TOutput>::Demand(TKey key)
		{
			return inputs_[key].Request();
		}

		template<typename TKey, typename TInput, typename TOutput>
		Selector<TKey, TInput, TOutput>::~Selector()
		{

		}
		//---------------------------------------------------------------------------------
		template<typename TKey, typename TInput, typename TOutput>
		class SimpleSelector : public Selector < TKey, TInput, TOutput > {
		public:
			SimpleSelector(TKey key1, TKey key2){ Add(key1); Add(key2); }
			SimpleSelector(TKey key1, TKey key2, TKey key3){ Add(key1); Add(key2); Add(key3); }
			SimpleSelector(TKey key1, TKey key2, TKey key3, TKey key4){ Add(key1); Add(key2); Add(key3); Add(key4); }
			SimpleSelector(TKey key1, TKey key2, TKey key3, TKey key4, TKey key5){ Add(key1); Add(key2); Add(key3); Add(key4); Add(key5); }
			SimpleSelector& operator=(const std::function<TOutput&(TInput&)>& process) { process_ = process; return *this; }
            void operator=(const TKey& key) { key_ = key; }
            Plug<TInput>& operator()(const TKey& key) { return Inputs(key); }
			virtual ~SimpleSelector(){}
		protected:
			SimpleSelector(SimpleSelector<TKey, TInput, TOutput>& peer) { }
			virtual TOutput& Process() override { TInput& value = Demand(key_);  return process_(value); }
		private:
			TKey key_;
			std::function<TOutput&(TInput&)> process_;
		};

		//---------------------------------------------------------------------------------

		template<typename TKey, typename TInput, typename TOutput> class Restrictor {
		public:
			Restrictor();
			virtual ~Restrictor();
			Plug<TInput>& Input(){ return input_; }
			KeyedSocket<TKey, TOutput>& Outputs(TKey key){ return outputs_.at(key); }
		protected:
			Restrictor(Restrictor<TKey, TInput, TOutput>& peer) { }
			virtual void Add(TKey key);
			virtual void Remove(TKey key);
			inline TInput& Demand() { return input_.Request(); }
			virtual TOutput& Process(TKey key) = 0;
		private:
			std::map<TKey, KeyedSocket<TKey, TOutput>> outputs_;
			Plug<TInput> input_;
		};

		template<typename TKey, typename TInput, typename TOutput>
		Restrictor<TKey, TInput, TOutput>::Restrictor()
		{
			
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Restrictor<TKey, TInput, TOutput>::Add(TKey key)
		{
			outputs_.emplace(key, key);
			outputs_.at(key).Bind(std::bind(&Restrictor<TKey, TInput, TOutput>::Process, this, std::placeholders::_1));
		}

		template<typename TKey, typename TInput, typename TOutput>
		void Restrictor<TKey, TInput, TOutput>::Remove(TKey key)
		{
			outputs_.erase(key);
		}

		template<typename TKey, typename TInput, typename TOutput>
		Restrictor<TKey, TInput, TOutput>::~Restrictor()
		{

		}

		//---------------------------------------------------------------------------------
		template<typename TKey, typename TInput, typename TOutput>
		class SimpleRestrictor : public Restrictor < TKey, TInput, TOutput > {
		public:
			SimpleRestrictor(TKey key1, TKey key2) { Add(key1); Add(key2); }
			SimpleRestrictor(TKey key1, TKey key2, TKey key3){ Add(key1); Add(key2); Add(key3); }
			SimpleRestrictor(TKey key1, TKey key2, TKey key3, TKey key4){ Add(key1); Add(key2); Add(key3); Add(key4); }
			SimpleRestrictor(TKey key1, TKey key2, TKey key3, TKey key4, TKey key5){ Add(key1); Add(key2); Add(key3); Add(key4); Add(key5); }
			SimpleRestrictor<TKey, TInput, TOutput>& operator=(const std::function<TOutput&(TKey, TInput&)>& process) { process_ = process; return *this; }
            SimpleRestrictor<TKey, TInput, TOutput>& operator=(const TKey& key) { key_ = key; return *this;}
            inline void SetDefault(TOutput* value) { result_ = value; }
            KeyedSocket<TKey, TOutput>& operator()(const TKey& key) { return Outputs(key); }
			virtual ~SimpleRestrictor(){}
		protected:
			SimpleRestrictor(SimpleRestrictor<TKey, TInput, TOutput>& peer) { }
			virtual TOutput& Process(TKey key) override { if (key == key_) { result_ = &process_(key, Demand()); } return *result_; }
		private:
			TOutput *result_;
			TKey key_;
			std::function<TOutput&(TKey, TInput&)> process_;
		};

		//---------------------------------------------------------------------------------

		template<typename T> class Separator {
		public:
			Separator();
			virtual ~Separator();
			Plug<T>& Input(){ return input_; }
			Socket<T>& Output(){ return output_; }
			inline void Wait() { signal_.wait(); }
		protected:
			Separator(Separator<T>& peer) { }
			virtual T& Supply();
			inline T& Demand() { return input_.Request(); }
		private:
			Plug<T> input_;
			Socket<T> output_;
			std::future<T*> signal_;
			std::atomic_flag started_;
			std::mutex mutex_;
		};

		template<typename T>
		Separator<T>::Separator()
		{
			output_.Bind(std::bind(&Separator<T>::Supply, this));
			started_.clear();
		}

		template<typename T>
		T& Separator<T>::Supply()
		{
			std::lock_guard<std::mutex> lock(mutex_);
			if (!started_.test_and_set()){
				signal_ = std::async(std::launch::async, [&]()->T* { return &Demand(); });
			}
			signal_.wait();
			T* result = signal_.get();
			signal_ = std::async(std::launch::async, [&]() { std::lock_guard<std::mutex> lock(mutex_); return &Demand(); });
			return *result;
		}

		template<typename T>
		Separator<T>::~Separator()
		{

		}

	}//end of Pullflow namespace
	namespace as
	{
		template<typename T>
		class Relayer
		{
		public:
			Relayer();
			virtual ~Relayer();
			inline pl::Plug<T>& Input() { return input_; }
			inline ps::Plug<T>& Output() { return output_; }
			void Process();
		protected:
			Relayer(Relayer<T>& peer) { }
			inline T& Demand() { return input_.Request(); }
			inline void Send(T& value) { output_.Transmit(value); }
		private:
			pl::Plug<T> input_;
			ps::Plug<T> output_;
		};

		template<typename T>
		Relayer<T>::Relayer()
		{

		}

		template<typename T>
		Relayer<T>::~Relayer()
		{

		}

		template<typename T>
		void Relayer<T>::Process()
		{
			Send(Demand());
		}

		template<typename T>
		class Accumulator
		{
		public:
			Accumulator(int capacity = 0);
			~Accumulator();
			inline ps::Socket<T>& Input() { return input_; }
			inline pl::Socket<T>& Output() { return output_; }
		protected:
			Accumulator(Accumulator<T>& peer) { }
			virtual void Receive(T& value);
			virtual T& Supply();
			inline const int Size() { return queue_.size(); }
		private:
			ps::Socket<T> input_;
			pl::Socket<T> output_;
			std::condition_variable cv_;
			std::queue<T*> pointers_;
			const int capacity_;
			std::mutex mutex_;
		};

		template<typename T>
		Accumulator<T>::Accumulator(int capacity) : capacity_(capacity)
		{
			input_.Bind(std::bind(&Accumulator<T>::Receive, this, std::placeholders::_1));
			output_.Bind(std::bind(&Accumulator<T>::Supply, this));
		}


		template<typename T>
		Accumulator<T>::~Accumulator()
		{
			std::lock_guard<std::mutex> lock(mutex_);
		}

		template<typename T>
		void Accumulator<T>::Receive(T& value)
		{
			std::unique_lock<std::mutex> lock(mutex_);
			const int count = pointers_.size();
			if (capacity_ > 0 && count >= capacity_){
				cv_.wait(lock);
			}
			pointers_.push(&value);
			if (count == 0){
				cv_.notify_all();
			}
		}

		template<typename T>
		T& Accumulator<T>::Supply()
		{
			std::unique_lock<std::mutex> lock(mutex_);
			const int count = pointers_.size();
			if (count == 0){
				cv_.wait(lock);
			}
			T* result = pointers_.front();
			pointers_.pop();
			if (capacity_ > 0 && count >= capacity_){
				cv_.notify_one();
			}
			return *result;
		}

        //--------------------------------------------------------------------------------------

		template<typename T>
		class Provider
		{
		public:
			Provider();
			Provider(const std::function<T*(void)>& factory);
			~Provider();
			inline ps::Socket<T>& Input() { return input_; }
			inline pl::Socket<T>& Output() { return output_; }
		protected:
			Provider(Provider<T>& peer) { }
			virtual void Receive(T& value);
			virtual T& Supply();
			virtual void Clean(T* obj);
		private:
			std::map<T*, std::unique_ptr<T>> references_;
			ps::Socket<T> input_;
			pl::Socket<T> output_;
			std::function<T*(void)> factory_;
			std::queue<T*> available_;
			std::mutex mutex_;

		};
		template<typename T>
		Provider<T>::Provider() // : Provider([]()->T* {return new T();}) <- VS2012 :(
		{
			input_.Bind(std::bind(&Provider<T>::Receive, this, std::placeholders::_1));
			output_.Bind(std::bind(&Provider<T>::Supply, this));
			factory_ = []()->T* {return new T(); };
		}

		template<typename T>
		Provider<T>::Provider(const std::function<T*(void)>& factory)
		{
			input_.Bind(std::bind(&Provider<T>::Receive, this, std::placeholders::_1));
			output_.Bind(std::bind(&Provider<T>::Supply, this));
			factory_ = factory;
		}


		template<typename T>
		Provider<T>::~Provider()
		{
			std::unique_lock<std::mutex> lock(mutex_);
		}

		template<typename T>
		void Provider<T>::Clean(T* obj)
		{

		}

		template<typename T>
		void Provider<T>::Receive(T& value)
		{
			std::unique_lock<std::mutex> lock(mutex_);
			const int count = available_.size();
			T* key = &value;
			std::map<T*, std::unique_ptr<T>>::iterator it = references_.find(key);
			if (it == references_.end()){
				key = new T(std::forward<T>(value));
				references_[key] = std::unique_ptr<T>(key);
			}
			Clean(key);
			available_.push(key);
		}

		template<typename T>
		T& Provider<T>::Supply()
		{
			std::unique_lock<std::mutex> lock(mutex_);
			T* result;
			const int count = available_.size();
			if (count == 0){
				result = new T();
				references_[result] = std::unique_ptr<T>(result);
				Clean(result);
				return *result;
			}
			result = available_.front();
			available_.pop();
			return *result;
		}

		template<typename T>
		class pl_ref_transceiver : public pl::Transceiver<T, T>
		{
		public:
			pl_ref_transceiver();
			pl_ref_transceiver(const std::function<T*(T&)>& factory);
			virtual ~pl_ref_transceiver();
			inline ps::Socket<int>& Clean() { return clean_; }
		protected:
			pl_ref_transceiver(pl_ref_transceiver<T>& peer) { }
			virtual T& Process() override;
			virtual int& Cleaning();
		private:
			pl::Socket<int> clean_;
			std::queue<std::unique_ptr<T>> pointers_;
			std::function<T*(T&)> factory_;
			int clean_signal_;
		};

		template<typename T>
		pl_ref_transceiver<T>::pl_ref_transceiver() : pl::Transceiver<T, T>(), clean_signal_(1)
		{
			factory_ = [](T& source)->T* { return new T(source); };
			clean_.Bind(std::bind(&pl_ref_transceiver<T>::Cleaning, this));
		}

		template<typename T>
		pl_ref_transceiver<T>::pl_ref_transceiver(const std::function<T*(T&)>& factory) : pl::Transceiver<T, T>(), factory_(factory), clean_signal_(1)
		{
			clean_.Bind(std::bind(&pl_ref_transceiver<T>::Cleaning, this));
		}

		template<typename T>
		pl_ref_transceiver<T>::~pl_ref_transceiver()
		{

		}

		template<typename T>
		T& pl_ref_transceiver<T>::Process()
		{
			T& source = Demand();
			T* result = factory_(source);
			pointers_.emplace(result);
			return *pointers_.front().get();
		}

		template<typename T>
		int& pl_ref_transceiver<T>::Cleaning()
		{
			while (!pointers_.empty()){
				pointers_.pop();
			}
			return clean_signal_;
		}

		template<typename T>
		class ps_ref_transceiver : public ps::Transceiver<T, T>
		{
		public:
			ps_ref_transceiver();
			ps_ref_transceiver(const std::function<T*(T&)>& factory);
			virtual ~ps_ref_transceiver();
			inline ps::Socket<int>& Clean() { return clean_; }
		protected:
			ps_ref_transceiver(ps_ref_transceiver<T>& peer) { }
			virtual void Process(T&) override;
			virtual void Cleaning(int&);
		private:
			ps::Socket<int> clean_;
			std::queue<std::unique_ptr<T>> pointers_;
			std::function<T*(T&)> factory_;
		};

		template<typename T>
		ps_ref_transceiver<T>::ps_ref_transceiver() : ps::Transceiver<T, T>()
		{
			factory_ = [](T& source)->T* { return new T(source); };
			clean_.Bind([&](int&v) {Cleaning(v); });
		}

		template<typename T>
		ps_ref_transceiver<T>::ps_ref_transceiver(const std::function<T*(T&)>& factory) : ps::Transceiver<T, T>(), factory_(factory)
		{
			clean_.Bind([&](int&v) {Cleaning(v); });
		}

		template<typename T>
		ps_ref_transceiver<T>::~ps_ref_transceiver()
		{

		}

		template<typename T>
		void ps_ref_transceiver<T>::Process(T& source)
		{
			T* result = factory_(source);
			pointers_.emplace(result);
			result = pointers_.front().get();
			Send(*result);
		}

		template<typename T>
		void ps_ref_transceiver<T>::Cleaning(int& v)
		{
			while (!pointers_.empty()){
				pointers_.pop();
			}
			v = 1;
		}
	}//end of as alias adapters namespace
}//end of dvo alias devlego namespace

#undef DISALLOW_ASSIGN
#undef DISALLOW_COPY
#undef DISALLOW_MOVE
#undef DISALLOW_COPY_AND_ASSIGN
#undef DISALLOW_MOVE_AND_ASSIGN
#undef DISALLOW_COPY_AND_MOVE_AND_ASSIGN